package br.unifor.g2cl;

import java.net.SocketAddress;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.Vector;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class RspList implements Map<SocketAddress, Rsp> {

	private ReentrantLock LOCK = new ReentrantLock();
	private Condition cond = LOCK.newCondition();
	
	private int requestId;
	private int requestMode;
	private int responsesReceived;
	private int waitNResposes;
	private int qtResponses;
	private boolean canReturn;
	// XXX não esta sendo usado
	public boolean suspected;
	private RspFilter filter = null;

	private Map<SocketAddress, Rsp> responses = new HashMap<SocketAddress, Rsp>();

	public RspList() {
		this.suspected = false;
		this.canReturn = false;
	}

	public RspList(int requestId, int requestMode) {
		this(requestId,requestMode,null);
	}
	
	public RspList(int requestId, int requestMode,RspFilter filter) {
		responses = new HashMap<SocketAddress, Rsp>();
		this.requestId = requestId;
		this.requestMode = requestMode;
		responsesReceived = 0;
		qtResponses = 0;
		this.filter = filter;
		
		if (requestMode == GroupRequest.GET_NONE) {
			canReturn = true;
			waitNResposes = 0;
		} else if (requestMode == GroupRequest.GET_FIRST) {
			canReturn = false;
			waitNResposes = 1;
		} else {
			canReturn = false;
		}
	}
	

	public void addRsp(SocketAddress address) {
		Rsp response = new Rsp(address);
		addRsp(address, response);
	}
	
	public void addRsps(List<SocketAddress> addresses) {
		for(SocketAddress address:addresses){
			Rsp response = new Rsp(address);
			addRsp(address, response);			
		}
	}
	
	

	public Rsp addRsp(SocketAddress address, Rsp response) {
		qtResponses++;
		if (requestMode == GroupRequest.GET_ALL) {
			waitNResposes = qtResponses;
		} else if (requestMode == GroupRequest.GET_MAJORITY) {
			waitNResposes = qtResponses / 2 + 1;
		}
		return responses.put(address, response);
	}

	public void receiveResponse(SocketAddress address, Object returnVal) {

		if (filter != null && !filter.isAcceptable(returnVal, address)) {
			LOCK.lock();
			try {
				if (!filter.needMoreResponses()) {
					cond.signalAll();
				}
			} finally {
				LOCK.unlock();
			}
			return;
		}

		Rsp response = responses.get(address);
		if (response != null) {
			response.setReceived(true);
			response.setValue(returnVal);
			responsesReceived++;
			LOCK.lock();
			try {
				canReturn = (responsesReceived >= waitNResposes);
				if (canReturn || (filter != null && !filter.needMoreResponses())) {
					cond.signalAll();
				}
			} finally {
				LOCK.unlock();
			}
		}
	}
	

	public void clear() {
		responses.clear();

	}

	public boolean containsKey(Object key) {
		return responses.containsKey(key);
	}

	public boolean containsValue(Object value) {
		return responses.containsValue(value);
	}

	public Set<java.util.Map.Entry<SocketAddress, Rsp>> entrySet() {
		return responses.entrySet();
	}

	public Rsp get(Object key) {
		return responses.get(key);
	}

	public boolean isEmpty() {
		return responses.isEmpty();
	}

	public Set<SocketAddress> keySet() {
		return responses.keySet();
	}

	public Rsp put(SocketAddress key, Rsp value) {
		return addRsp(key, value);
	}

	public void putAll(Map<? extends SocketAddress, ? extends Rsp> t) {
		responses.putAll(t);
	}

	public Rsp remove(Object key) {
		return responses.remove(key);
	}

	public int size() {
		return responses.size();
	}

	public Collection<Rsp> values() {
		return responses.values();
	}

	//TODO------------------------------

	/** Adds a list of responses
	 * @param responses Collection<Rsp>
	 */
	public RspList(Collection<Rsp> responses) {
		if (responses != null) {
			for (Rsp rsp : responses) {
				put(rsp.getSender(), rsp);
			}
		}
	}

	/**
	 * Returns the value associated with address key
	 * @param key
	 * @return Object value
	 */
	public Object getValue(Object key) {
		Rsp rsp = get(key);
		return rsp != null ? rsp.getValue() : null;
	}

	/**
	 * Clears the response list
	 * @deprecated Use {@link #clear()} instead
	 */
	public void reset() {
		clear();
	}

	public boolean isReceived(SocketAddress sender) {
		Rsp rsp = get(sender);
		return rsp != null && rsp.received;
	}

	public int numSuspectedMembers() {
		int num = 0;
		Collection<Rsp> values = values();
		for (Rsp rsp : values) {
			if (rsp.wasSuspected())
				num++;
		}
		return num;
	}

	public int numReceived() {
		int num = 0;
		for (Rsp rsp : values()) {
			if (rsp.wasReceived())
				num++;
		}
		return num;
	}

	/** Returns the first value in the response set. This is random, but we try to return a non-null value first */
	public Object getFirst() {

		for (Rsp rsp : values()) {
			if (rsp.getValue() != null)
				return rsp.getValue();
		}
		return null;
	}

	/**
	 * Returns the results from non-suspected members that are not null.
	 */
	public Vector<Object> getResults() {
		Vector<Object> ret = new Vector<Object>();
		Object val;

		for (Rsp rsp : values()) {
			if (rsp.wasReceived() && (val = rsp.getValue()) != null)
				ret.addElement(val);
		}
		return ret;
	}

	public Vector<SocketAddress> getSuspectedMembers() {
		Vector<SocketAddress> retval = new Vector<SocketAddress>();
		for (Rsp rsp : values()) {
			if (rsp.wasSuspected())
				retval.addElement(rsp.getSender());
		}
		return retval;
	}

	public boolean isSuspected(SocketAddress sender) {
		Rsp rsp = get(sender);
		return rsp != null && rsp.suspected;
	}

	public String toString() {
		StringBuilder ret = new StringBuilder();
		for (Rsp rsp : values()) {
			ret.append("[" + rsp + "]\n");
		}
		return ret.toString();
	}

	boolean contains(SocketAddress sender) {
		return containsKey(sender);
	}

	public void addSuspect(SocketAddress sender) {
		Rsp rsp = get(sender);

		if (rsp != null && !rsp.received) {
			// Recalculating the number of responses to wait!
			qtResponses--;
			if (requestMode == GroupRequest.GET_ALL) {
				waitNResposes = qtResponses;
			} else if (requestMode == GroupRequest.GET_MAJORITY) {
				waitNResposes = qtResponses / 2 + 1;
			} else if (requestMode == GroupRequest.GET_FIRST
					&& qtResponses == 0) {
				waitNResposes = 0;					
			}

			rsp.sender = sender;
			rsp.retval = null;
			rsp.received = false;
			rsp.suspected = true;

			LOCK.lock();
			try {
				if (!canReturn) {
					canReturn = responsesReceived >= waitNResposes;
					if (canReturn == true) {
						cond.signalAll();	
					}
				}
			} finally {
				LOCK.unlock();
			}
		}
	}

	
	
	
	
	/**
	 * Returns the MessageDispatcherResponse at index i
	 * @param i The index
	 * @return a MessageDispatcherResponse
	 * @throws ArrayIndexOutOfBoundsException
	 * @deprecated Use {@link #entrySet()} or {@link #values()} instead
	 */
	public Object elementAt(int i) throws ArrayIndexOutOfBoundsException {
		Set<SocketAddress> keys = new TreeSet<SocketAddress>(
				new SocketAddressComparator());
		keys.addAll(keySet());
		Object[] keys_array = keys.toArray();
		Object key = keys_array[i];
		return get(key);
	}

	class SocketAddressComparator implements Comparator<SocketAddress> {
		public int compare(SocketAddress o1, SocketAddress o2) {
			return new Integer(o1.hashCode()).compareTo(new Integer(o2
					.hashCode()));
		}
	}

	
	
	public void waitResponses(long timeout) {
		LOCK.lock();
		if (canReturn) {
			LOCK.unlock();
			return;
		}
		try {
			if (timeout > 0) {
				cond.await(timeout, TimeUnit.MILLISECONDS);
			} else {
				cond.await();
			}

		} catch (InterruptedException e) {
			e.printStackTrace();

		} finally {
			LOCK.unlock();
		}
	}

}
